package batch;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

import java.sql.SQLOutput;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class operator_map {
    public static void main(String[] args) throws Exception {
    //    创建运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    //    获取数据
        ArrayList<String> data = new ArrayList<String>();
        data.add("i love bing");
        data.add("i love shui");
        data.add("bing shui");
        DataSet<String> source =  env.fromCollection(data);// data 可以是个list
        //System.out.println(source.getClass().toString());
        System.out.println("*****map****");
    //    每句话进行处理，得到含有单词的list中，List<String>是返回的List
        DataSet<List<String>> result1 = source.map(new MapFunction<String, List<String>>() {
            //用实现map 函数
            public List<String> map(String value) throws Exception {
                // 这里的value是每句话，也就是i love bing，i love shui，bing shui
                String[] words = value.split(" ");

                //ArrayList<String> result = new ArrayList<String>();
                ////System.out.println(value + value.getClass().toString());
                //for(String word: words){
                //    result.add(word);
                //}
                List<String> result = Arrays.asList(words);
                //ArrayList<String> result = Arrays.asList(words);

                return result;
            }
        });
        result1.print();
        System.out.println("*****map****");

    }
}
